package com.winit.schedule.task.tracker.runner;

import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.winit.schedule.core.model.AssignedTask;
import com.winit.schedule.core.model.Task;
import com.winit.schedule.core.model.TaskItem;
import com.winit.schedule.core.repository.constants.TaskStatus;
import com.winit.schedule.core.repository.entity.TaskItemEntity;
import com.winit.schedule.core.task.ScheduleTaskDealMulti;
import com.winit.schedule.core.task.annotation.ScheduleParam;
import com.winit.schedule.core.utils.Clock;
import com.winit.schedule.task.tracker.utils.ExceptionUtils;

/**
 * 多线程任务执行器
 * 
 * @author jianke.zhang 2015-4-22 上午9:57:00
 * @since 1.0
 */
public class MultiThreadTaskRunner<T> implements TaskRunner, Runnable {

    private static transient Logger    logger              = LoggerFactory.getLogger(MultiThreadTaskRunner.class);

    /**
     * 运行线程
     */
    List<Thread>                       threadList          = Collections.synchronizedList(new ArrayList<Thread>());

    /**
     * 任务数据提取
     */
    protected List<T>                  subTaskList         = Collections.synchronizedList(new ArrayList<T>());

    private final ReentrantLock        runLock             = new ReentrantLock();

    /**
     * 任务管理器
     */
    protected AbstractTaskManager      taskManager;

    /**
     * 任务
     */
    protected AssignedTask             assignedTask;

    /**
     * 传递给任务执行接口的参数
     */
    private Task                       task;

    /**
     * 任务分析
     */
    protected StatisticsInfo           statisticsInfo;

    /**
     * 队列处理器
     */
    protected ScheduleTaskDealMulti<T> taskDealBean;

    /**
     * 线程是否处于休眠状态
     */
    boolean                            isSleeping          = false;

    /**
     * 异常或人工终止
     */
    boolean                            isStop              = false;

    /**
     * 任务完成
     */
    boolean                            isCompleteSchedule  = false;

    /**
     * 线程调度锁对象
     */
    final ThreadScheduleLockObject     ts_lockObject       = new ThreadScheduleLockObject();

    /**
     * 单次调度抓取数据次数；0表示无限制，大于0表示实际限制次数
     */
    int                                limitFetchDataTimes = 0;

    volatile int                       fetchDataTimes      = 0;

    public MultiThreadTaskRunner(AbstractTaskManager taskManager, ScheduleTaskDealMulti<T> taskDealBean,
                                 StatisticsInfo statisticsInfo){
        this.taskManager = taskManager;
        this.taskDealBean = taskDealBean;
        this.statisticsInfo = statisticsInfo;
        this.assignedTask = taskManager.getTask();

        // 初始化单次调度抓取任务的次数
        ScheduleParam scheduleParam = taskDealBean.getClass().getAnnotation(ScheduleParam.class);
        if (null != scheduleParam && scheduleParam.limitFetchDataTimes() > 0) {
            limitFetchDataTimes = scheduleParam.limitFetchDataTimes();
        } else {
            limitFetchDataTimes = 0; // 无限制
        }

        // 构建任务接口接收参数
        this.task = buildTask(this.assignedTask);

        if (assignedTask.getEachFetchDataNum() < assignedTask.getThreadsOfRegion() * 10) {
            logger.warn("参数设置不合理，系统性能不佳。【每次从数据库获取的数量fetchnum】 >= 【线程数量threadnum】 *【最少循环次数10】 ");
        }

        if (assignedTask.getThreadsOfRegion() < 1) {
            logger.warn("参数设置错误，执行定时任务开启的线程必须大于等于1.");
            assignedTask.setThreadsOfRegion(1);
        }

        if (assignedTask.getEachFetchDataNum() < assignedTask.getThreadsOfRegion()) {
            assignedTask.setEachFetchDataNum(assignedTask.getThreadsOfRegion());
        }

        for (int i = 0; i < assignedTask.getThreadsOfRegion(); i++) {
            this.startThread(i);
        }
    }

    private Task buildTask(AssignedTask assignedTask) {
        Task task = new Task();
        task.setEachFetchDataNum(assignedTask.getEachFetchDataNum());
        task.setTaskParameter(assignedTask.getTaskParameter());

        List<TaskItem> taskItems = new ArrayList<TaskItem>();
        if (assignedTask.getTaskItems() != null) {
            for (TaskItemEntity item : assignedTask.getTaskItems()) {
                taskItems.add(new TaskItem(item.getTaskItemId(), item.getParameter()));
            }
            task.setTaskItems(taskItems);
        }
        return task;
    }

    public void startThread(int index) {
        Thread thread = new Thread(this);
        threadList.add(thread);
        String threadName = this.taskManager.getTask().getTaskId() + "-" + this.taskManager.getCurrentSerialNumber()
                            + "-exe-" + index;
        thread.setName(threadName);
        thread.start();
    }

    @Override
    public boolean isDealFinishAllData() {
        return this.subTaskList.size() == 0;
    }

    @Override
    public boolean isSleeping() {
        return this.isSleeping;
    }

    @Override
    public void completeSchedule() throws Exception {
        this.isCompleteSchedule = true;
        this.stopTask(false);
    }

    @Override
    public void stopTask(boolean force) throws Exception {
        this.isStop = true;

        if (force && runLock.tryLock()) {
            try {
                for (Thread thread : threadList) {
                    try {
                        if (!thread.isInterrupted()) {
                            thread.interrupt();
                        }
                    } catch (Throwable e) {

                    }
                }
            } finally {
                runLock.unlock();
            }
        }

        // 清除所有未处理任务,但已经进入处理队列的，需要处理完毕
        this.subTaskList.clear();

    }

    @Override
    public void clearAllHasFetchData() {
        this.subTaskList.clear();
    }

    public synchronized T[] getScheduleTaskMulti() {

        if (this.subTaskList.size() == 0) {
            return null;
        }
        int size = subTaskList.size() > assignedTask.getBatchDealNum() ? assignedTask.getBatchDealNum() : subTaskList.size();

        T[] result = null;
        if (size > 0) {
            result = (T[]) Array.newInstance(this.subTaskList.get(0).getClass(), size);
        }
        for (int i = 0; i < size; i++) {
            result[i] = this.subTaskList.remove(0);
        }
        return result;
    }

    protected int loadScheduleData() {
        try {
            // 在每次数据处理完毕后休眠固定的时间
            if (this.assignedTask.getSleepTimeInterval() > 0) {
                if (logger.isTraceEnabled()) {
                    logger.trace("Finished processing a batch of data after dormancy："
                                 + this.assignedTask.getSleepTimeInterval());
                }
                this.isSleeping = true;
                Thread.sleep(assignedTask.getSleepTimeInterval());
                this.isSleeping = false;

                if (logger.isTraceEnabled()) {
                    logger.trace("Finished processing a number of data recovery after dormancy");
                }
            }

            // 超过最大抓取次数限制则不再抓取数据
            if (limitFetchDataTimes > 0 && ++fetchDataTimes > limitFetchDataTimes) {
                return 0;
            }

            List<TaskItemEntity> taskItems = this.assignedTask.getTaskItems();
            // 根据队列信息查询需要调度的数据，然后增加到任务列表中
            if (taskItems.size() > 0) {
                List<T> tmpList = this.taskDealBean.selectTasks(this.task);
                this.statisticsInfo.setLastFetchDataTime(Clock.getSystemTime());
                if (tmpList != null) {
                    this.subTaskList.addAll(tmpList);
                }
            } else {
                if (logger.isTraceEnabled()) {
                    logger.trace("There is no access to the need to deal with the data queue");
                }
            }
            addFetchNum(subTaskList.size(), "DefaultTaskRunner.loadScheduleData");
            return this.subTaskList.size();
        } catch (Throwable ex) {
            logger.error("Get tasks error.", ex);
        }
        return 0;
    }

    @SuppressWarnings("static-access")
    @Override
    public void run() {

        try {
            long startTime = 0;
            while (true) {
                this.ts_lockObject.addThread();
                T[] executeTask;
                while (true) {
                    if (this.isStop == true) {// 停止队列调度
                        this.ts_lockObject.realseThread();
                        this.ts_lockObject.notifyOtherThread();// 通知所有的休眠线程
                        synchronized (this.threadList) {
                            this.threadList.remove(Thread.currentThread());
                            if (this.threadList.size() == 0) {
                                // 任务结束
                                if (isCompleteSchedule) {
                                    this.taskManager.completeTask(TaskStatus.COMPLETE, null);
                                } else {
                                    this.taskManager.completeTask(TaskStatus.STOP, null);
                                }
                            }
                        }
                        return;
                    }

                    executeTask = this.getScheduleTaskMulti();

                    if (executeTask == null) {
                        break;
                    }

                    try {// 运行相关的程序
                        startTime = Clock.getSystemTime();
                        if (this.taskDealBean.execute(executeTask) == true) {
                            addSuccessNum(1,
                                Clock.getSystemTime() - startTime,
                                "com.winit.schedule.task.tracker.runner.DefaultTaskRunner.run");
                        } else {
                            addFailNum(1,
                                Clock.getSystemTime() - startTime,
                                "com.winit.schedule.task.tracker.runner.DefaultTaskRunner.run");
                            this.taskManager.fireTaskExecuteFailure(null);
                        }
                    } catch (Throwable ex) {
                        addFailNum(1, Clock.getSystemTime() - startTime, "DefaultTaskRunner.run");
                        logger.warn("Task :" + executeTask + " failure", ex);
                        this.taskManager.fireTaskExecuteFailure(ExceptionUtils.getStackTrace(ex));
                    }
                }
                // 当前队列中所有的任务都已经完成了。
                if (logger.isTraceEnabled()) {
                    logger.trace(Thread.currentThread().getName() + "：The currently running thread number:"
                                 + this.ts_lockObject.count());
                }
                if (this.ts_lockObject.realseThreadButNotLast() == false) {
                    int size = 0;
                    Thread.currentThread().sleep(100L);
                    startTime = Clock.getSystemTime();
                    // 装载数据
                    size = this.loadScheduleData();
                    if (size > 0) {
                        this.ts_lockObject.notifyOtherThread();
                    } else {
                        // 判断当没有数据的是否，是否需要退出调度
                        if (this.isStop == false && this.taskManager.isContinueWhenNoData() == true) {
                            if (logger.isTraceEnabled()) {
                                logger.trace("No loading into the data,start sleep.");
                            }
                            this.isSleeping = true;
                            Thread.currentThread().sleep(this.assignedTask.getSleepTimeNoData());
                            this.isSleeping = false;

                            if (logger.isTraceEnabled()) {
                                logger.trace("Sleep end");
                            }
                        } else {
                            // 没有数据，退出调度，唤醒所有沉睡线程
                            this.ts_lockObject.notifyOtherThread();
                        }
                    }
                    this.ts_lockObject.realseThread();
                } else {// 将当前线程放置到等待队列中。直到有线程装载到了新的任务数据
                    if (logger.isTraceEnabled()) {
                        logger.trace("Current thread not last running thread，go to sleep");
                    }
                    this.ts_lockObject.waitCurrentThread();
                }
            }
        } catch (InterruptedException ex) {
            logger.warn("Task :" + assignedTask + " failure", ex);
            this.taskManager.completeTask(TaskStatus.STOP, null);
        } catch (Throwable e) {
            logger.error(e.getMessage(), e);
            this.taskManager.completeTask(TaskStatus.RUNNER_FAILURE, ExceptionUtils.getStackTrace(e));
        }
    }

    public int getThreadNum() {
        return this.threadList.size();
    }

    public void addFetchNum(long num, String addr) {

        this.statisticsInfo.addFetchDataCount(1);
        this.statisticsInfo.addFetchDataNum(num);
    }

    public void addSuccessNum(long num, long spendTime, String addr) {
        this.statisticsInfo.addDealDataSucess(num);
        this.statisticsInfo.addDealSpendTime(spendTime);
    }

    public void addFailNum(long num, long spendTime, String addr) {
        this.statisticsInfo.addDealDataFail(num);
        this.statisticsInfo.addDealSpendTime(spendTime);
    }

}
